package svc

import (
	"github.com/apache/pulsar-client-go/pulsar"
	"github.com/minio/minio-go/v7"
	"github.com/minio/minio-go/v7/pkg/credentials"
	"github.com/nsqio/go-nsq"
	"github.com/zeromicro/go-queue/kq"
	"github.com/zeromicro/go-zero/zrpc"
	"my-zero-tiktok/apps/api/internal/config"
	"my-zero-tiktok/apps/comment/rpc/comment"
	"my-zero-tiktok/apps/message/rpc/message"
	"my-zero-tiktok/apps/user/rpc/user"
	"my-zero-tiktok/apps/video/rpc/video"
)

type ServiceContext struct {
	Config     config.Config
	UserRPC    user.UserZrpcClient
	VideoRPC   video.Video
	CommentRPC comment.CommentZrpcClient
	MessageRPC message.MessageZrpcClient
	OSSClient  *minio.Client

	PulsarClient   pulsar.Client
	PulsarConsumer pulsar.Consumer
	PulsarProducer pulsar.Producer

	KqProducer *kq.Pusher

	NSQProducer *nsq.Producer
	NSQConsumer *nsq.Consumer
}

type Handler struct{}

func NewServiceContext(c config.Config) *ServiceContext {
	minioClient, err := minio.New(
		c.OSS.EndPoint,
		&minio.Options{Creds: credentials.NewStaticV4(c.OSS.AccessKey, c.OSS.AccessSecret, ""), Secure: c.OSS.SSL},
	)
	if err != nil {
		panic("Failed to create minio client: " + err.Error())
	}

	// 创建 Pulsar 客户端
	client, err := pulsar.NewClient(pulsar.ClientOptions{
		URL: c.Pulsar.URL,
	})
	if err != nil {
		panic("Failed to create client: " + err.Error())
	}

	// 创建生产者
	producer, err := client.CreateProducer(pulsar.ProducerOptions{
		Topic: c.Pulsar.Topic,
	})
	if err != nil {
		panic("Failed to create producer: " + err.Error())
	}

	// 创建消费者
	consumer, err := client.Subscribe(pulsar.ConsumerOptions{
		Topic:            c.Pulsar.Topic,
		SubscriptionName: "my-subscription",
		Type:             pulsar.Shared,
	})
	if err != nil {
		panic("Failed to create consumer: " + err.Error())
	}

	nsqConfig := nsq.NewConfig()
	nsqProducer, err := nsq.NewProducer(c.NSQ.URL, nsqConfig)
	if err != nil {
		panic("Failed to create producer: " + err.Error())
	}
	nsqConsumer, err := nsq.NewConsumer(c.NSQ.Topic, "chanel_001", nsqConfig)
	if err != nil {
		panic("Failed to create consumer: " + err.Error())
	}
	nsqConsumer.AddHandler(&Handler{})
	err = nsqConsumer.ConnectToNSQD(c.NSQ.URL)
	if err != nil {
		panic("Failed to connect to NSQ: " + err.Error())
	}

	return &ServiceContext{
		Config:         c,
		UserRPC:        user.NewUserZrpcClient(zrpc.MustNewClient(c.UserRPC)),
		VideoRPC:       video.NewVideo(zrpc.MustNewClient(c.VideoRPC)),
		CommentRPC:     comment.NewCommentZrpcClient(zrpc.MustNewClient(c.CommentRPC)),
		MessageRPC:     message.NewMessageZrpcClient(zrpc.MustNewClient(c.MessageRPC)),
		OSSClient:      minioClient,
		PulsarClient:   client,
		PulsarProducer: producer,
		PulsarConsumer: consumer,
		KqProducer:     kq.NewPusher(c.Kafka.Brokers, c.Kafka.Topic),
		NSQProducer:    nsqProducer,
		NSQConsumer:    nsqConsumer,
	}
}
